package com.lhh.algorithmservice.service.Impl;

import com.lhh.algorithmservice.config.SparkConfigurationProperties;
import com.lhh.algorithmservice.service.CalculateRatingService;
import com.lhh.springcommon.annotation.WebLog;
import com.lhh.springcommon.annotation.WithRequestId;
import lombok.extern.slf4j.Slf4j;
import org.apache.spark.ml.evaluation.RegressionEvaluator;
import org.apache.spark.ml.recommendation.ALS;
import org.apache.spark.ml.recommendation.ALSModel;
import org.apache.spark.sql.*;
import org.apache.spark.sql.types.DataTypes;
import org.apache.spark.sql.types.Metadata;
import org.apache.spark.sql.types.StructField;
import org.apache.spark.sql.types.StructType;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import scala.collection.immutable.Seq;

import java.io.IOException;
import java.lang.reflect.UndeclaredThrowableException;
import java.util.Collections;
import java.util.List;
import java.util.Properties;
import java.util.stream.Collectors;

/**
 * @author 李厚航
 * @date 2025-03-15
 */
@Service
@Slf4j
public class CalculateRatingServiceImpl implements CalculateRatingService {


    private final SparkSession sparkSession;

    private final SparkConfigurationProperties sparkConfigurationProperties;

    @Autowired
    public CalculateRatingServiceImpl(SparkSession sparkSession,
                                      SparkConfigurationProperties sparkConfigurationProperties) {
        this.sparkSession = sparkSession;
        this.sparkConfigurationProperties = sparkConfigurationProperties;
    }

    @Override
    public void generateRatingCsv() {
        Properties properties = new Properties();
        properties.put("driver", sparkConfigurationProperties.getDataSource().getDriverClassName());
        properties.put("user", sparkConfigurationProperties.getDataSource().getUsername());
        properties.put("password", sparkConfigurationProperties.getDataSource().getPassword());
        Dataset<Row> userRow = sparkSession.read().jdbc(sparkConfigurationProperties.getDataSource().getUrl(), "user", properties);
        Dataset<Row> videoRow = sparkSession.read().jdbc(sparkConfigurationProperties.getDataSource().getUrl(), "video", properties);
        Dataset<Row> collectRow = sparkSession.read().jdbc(sparkConfigurationProperties.getDataSource().getUrl(), "collect", properties);
        Dataset<Row> commentRow = sparkSession.read().jdbc(sparkConfigurationProperties.getDataSource().getUrl(), "comment", properties);
        Dataset<Row> historyRow = sparkSession.read().jdbc(sparkConfigurationProperties.getDataSource().getUrl(), "history", properties);
        Dataset<Row> likeRow = sparkSession.read().jdbc(sparkConfigurationProperties.getDataSource().getUrl(), "`like`", properties);
        // 注册临时视图
        userRow.createOrReplaceTempView("user");
        videoRow.createOrReplaceTempView("video");
        collectRow.createOrReplaceTempView("collect");
        commentRow.createOrReplaceTempView("comment");
        historyRow.createOrReplaceTempView("history");
        likeRow.createOrReplaceTempView("like");
        // 多表查询：计算每个用户和视频的互动分数
        String sqlQuery =
                "SELECT " +
                        "   u.id AS uid, " +
                        "   v.id AS vid, " +
                        "   COALESCE(COUNT(h.id)*"+sparkConfigurationProperties.getUserVideoScoreWeight().getHistory()+", 0) + " +
                        "   COALESCE(COUNT(l.id)*"+sparkConfigurationProperties.getUserVideoScoreWeight().getLike()+", 0) + " +
                        "   COALESCE(COUNT(c.id)*"+sparkConfigurationProperties.getUserVideoScoreWeight().getCollect()+", 0) + " +
                        "   COALESCE(COUNT(cm.id)*"+sparkConfigurationProperties.getUserVideoScoreWeight().getComment()+", 0) AS score " +
                        "FROM user u " +
                        "CROSS JOIN video v " +  // 用户和视频的笛卡尔积
                        "LEFT JOIN `history` h ON u.id = h.uid AND v.id = h.vid " +  // 观看记录
                        "LEFT JOIN `like` l ON u.id = l.uid AND v.id = l.vid " +   // 点赞记录
                        "LEFT JOIN `collect` c ON u.id = c.uid AND v.id = c.vid " +  // 收藏记录
                        "LEFT JOIN `comment` cm ON u.id = cm.uid AND v.id = cm.vid " + // 评论记录
                        "GROUP BY u.id, v.id";
        Dataset<Row> var1 = sparkSession.sql(sqlQuery);
        DataFrameWriter<Row> var2 = var1.coalesce(2)
                .write()
                .option("charset", "UTF-8")
                .option("header", "true")
                .mode("overwrite");
        try{
            var2.csv(sparkConfigurationProperties.getScorePath());
        }catch (Exception e){
            e.printStackTrace();
        }
    }

    @Override
    public void trainALSModel() {
        // 1. 读取评分数据
        Dataset<Row> ratings = sparkSession.read()
                .format("csv")
                //第一行作为列名
                .option("header", "true")
                // 自动推断数据类型
                .option("inferSchema", "true")
                .load(sparkConfigurationProperties.getScorePath());

        // 2. 数据预处理,删除包含缺失值的行
        ratings.na().drop();

        // 3. 划分训练集和测试集
        Dataset<Row>[] splits = ratings.randomSplit(new double[]{0.8, 0.2}, 123L);
        Dataset<Row> training = splits[0];
        Dataset<Row> test = splits[1];

        // 4. 构建 ALS 模型
        ALS als = new ALS()
                // 最大迭代次数
                .setMaxIter(10)
                // 正则化参数
                .setRegParam(0.01)
                // 用户列
                .setUserCol("uid")
                // 视频列
                .setItemCol("vid")
                // 评分列
                .setRatingCol("score");

        // 5. 训练模型
        ALSModel model = als.fit(training);

        // 6. 评估模型
        Dataset<Row> predictions = model.transform(test);
        RegressionEvaluator evaluator = new RegressionEvaluator()
                // 使用 RMSE 作为评估指标
                .setMetricName("rmse")
                .setLabelCol("score")
                .setPredictionCol("prediction");
        double rmse = evaluator.evaluate(predictions);
        log.info("评估模型参数:{}",rmse);

        // 7. 保存模型
        try {
            model.write().overwrite().save(sparkConfigurationProperties.getModelPath());
        } catch (IOException e) {
            log.error("保存模型失败", e);
        }
    }

    @Override
    @WithRequestId
    @WebLog
    public List<Integer> recommendVideos(Integer userId, Integer numRecommendations) {

        //加载ALS模型
        ALSModel model = ALSModel.load(sparkConfigurationProperties.getModelPath());

        StructType schema = new StructType(new StructField[]{
                new StructField("uid", DataTypes.IntegerType, false, Metadata.empty())
        });

        // 创建包含用户 ID 的行
        Row row = RowFactory.create(userId);

        Dataset<Row> userDF = sparkSession.createDataFrame(Collections.singletonList(row), schema);

        Dataset<Row> userRecs = model.recommendForUserSubset(userDF, numRecommendations);

        return userRecs
                .select(functions.explode(functions.col("recommendations")).as("recommendation"))
                .select("recommendation.vid")
                // 将 vid 列转换为 Integer 类型
                .as(Encoders.INT())
                // 收集为 List<Integer>
                .collectAsList();

    }
}
